package org.pentaho.redis.output;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.*;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.params.SetParams;

import java.net.URI;
import java.net.URISyntaxException;


public class RedisStep extends BaseStep implements StepInterface {
    private static final Class<?> PKG = RedisStepMeta.class; // for i18n purposes

    private JedisPool jedisPool;
    private String redisKey;
    private Boolean redisKeyInField;
    private String redisKeyField;
    private String redisType;
    private String redisValueField;
    private Long millisecondsToExpire;

    /**
     * @param s                 step description
     * @param stepDataInterface step data class
     * @param c                 step copy
     * @param t                 transformation description
     * @param dis               transformation executing
     */
    public RedisStep(StepMeta s, StepDataInterface stepDataInterface, int c, TransMeta t, Trans dis) {
        super(s, stepDataInterface, c, t, dis);
    }

    /**
     * @param smi step meta interface implementation, containing the step settings
     * @param sdi step data interface implementation, used to store runtime information
     * @return true if initialization completed successfully, false if there was an error preventing the step from working.
     */
    public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
        // Casting to step-specific implementation classes is safe
        RedisStepMeta meta = (RedisStepMeta) smi;
        RedisStepData data = (RedisStepData) sdi;

        if (!super.init(meta, data)) {
            return false;
        }

        // 初始化 jedisPool
        try {
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            jedisPoolConfig.setMaxTotal(200);
            jedisPoolConfig.setMaxIdle(30);
            jedisPoolConfig.setMaxWaitMillis(60 * 1000);
            jedisPool = new JedisPool(jedisPoolConfig, new URI(((RedisStepMeta) smi).getRedisUrl()));
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }

        redisKey = meta.getRedisKey();
        redisKeyInField = meta.getRedisKeyInField();
        redisKeyField = meta.getRedisKeyField();
        redisType = meta.getRedisType();
        redisValueField = meta.getRedisValueField();
        millisecondsToExpire = meta.getMillisecondsToExpire();

        return true;
    }

    /**
     * @param smi the step meta interface containing the step settings
     * @param sdi the step data interface that should be used to store
     * @return true to indicate that the function should be called again, false if the step is done
     */
    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        // safely cast the step settings (meta) and runtime info (data) to specific implementations
        RedisStepMeta meta = (RedisStepMeta) smi;
        RedisStepData data = (RedisStepData) sdi;

        // get incoming row, getRow() potentially blocks waiting for more rows, returns null if no more rows expected
        Object[] r = getRow();

        // if no more rows are expected, indicate step is finished and processRow() should not be called again
        if (r == null) {
            setOutputDone();
            return false;
        }

        // the "first" flag is inherited from the base step implementation
        // it is used to guard some processing tasks, like figuring out field indexes
        // in the row structure that only need to be done once
        if (first) {
            first = false;
            data.outputRowMeta = (RowMetaInterface) getInputRowMeta().clone();

            if (redisKeyInField) {
                data.redisKeyFieldIndex = data.outputRowMeta.indexOfValue(redisKeyField);
                if (data.redisKeyFieldIndex < 0) {
                    throw new KettleException("找不到【" + redisKeyField + "】字段");
                }
            }

            data.redisValueFieldIndex = data.outputRowMeta.indexOfValue(redisValueField);
            if (data.redisValueFieldIndex < 0) {
                throw new KettleException("找不到【" + redisValueField + "】字段");
            }
        }

        String _redisKey;
        if (data.redisKeyFieldIndex < 0) {
            _redisKey = redisKey;
        } else {
            _redisKey = getInputRowMeta().getString(r, data.redisKeyFieldIndex);
        }

        Jedis jedis = jedisPool.getResource();
        switch (redisType) {
            case "String":
                if (millisecondsToExpire != 0) {
                    SetParams params = new SetParams();
                    params.px(millisecondsToExpire);
                    jedis.set(_redisKey, getInputRowMeta().getString(r, data.redisValueFieldIndex), params);
                } else {
                    jedis.set(_redisKey, getInputRowMeta().getString(r, data.redisValueFieldIndex));
                }
                break;
            case "List":
                jedis.rpush(_redisKey, getInputRowMeta().getString(r, data.redisValueFieldIndex));
                break;
            case "Set":
                jedis.sadd(_redisKey, getInputRowMeta().getString(r, data.redisValueFieldIndex));
                break;
        }
        jedis.close();

        // put the row to the output row stream
        putRow(data.outputRowMeta, r);

        // indicate that processRow() should be called again
        return true;
    }

    /**
     * @param smi step meta interface implementation, containing the step settings
     * @param sdi step data interface implementation, used to store runtime information
     */
    public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
        // Casting to step-specific implementation classes is safe
        RedisStepMeta meta = (RedisStepMeta) smi;
        RedisStepData data = (RedisStepData) sdi;

        // Add any step-specific initialization that may be needed here
        jedisPool.close();

        // Call superclass dispose()
        super.dispose(meta, data);
    }
}
